package cn.iocoder.springboot.lab31.rocketmqdemo.consumer;

import cn.hutool.bloomfilter.BitMapBloomFilter;
import cn.hutool.core.lang.UUID;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;


import java.util.List;


public class TestRepeatProducer {
    /**
     * 在boot项目中可以使用@Bean在整个容器中放置一个单利对象
     */
    public static BitMapBloomFilter bloomFilter = new BitMapBloomFilter(100); // m数组长度

    public void testRepeatConsumer() throws Exception {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("repeat-consumer-group");
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("repeatTestTopic", "*");
        // 注册一个消费监听 MessageListenerConcurrently是并发消费
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                // 拿到消息的key
                MessageExt messageExt = msgs.get(0);
                String keys = messageExt.getKeys();
                // 判断是否存在布隆过滤器中
                if (bloomFilter.contains(keys)) {
                    // 直接返回了 不往下处理业务
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                // 这个处理业务，然后放入过滤器中
                // do sth...
                bloomFilter.add(keys);
                System.out.println("keys:" + keys);
                System.out.println(new String(messageExt.getBody()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.in.read();
    }
}
